package vn.ebuzz.sms.worker;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.LinkedList;
import java.util.List;

import org.apache.log4j.Logger;

import vn.ebuzz.sms.core.Configuration;
import vn.ebuzz.sms.core.Database;
import vn.ebuzz.sms.core.Incoming;

public class IncomingQueue{
	private static Logger log = Logger.getLogger("inqueue");

	private static IncomingQueue q = null;
	private LinkedList<Incoming> queue = null;
	private boolean isRunning = true;

	private IncomingQueue(){
		queue = new LinkedList<Incoming>();
		for(int i = 0; i < Configuration.MAX_THREAD; i ++){
			IncomingWorker t = new IncomingWorker();
			t.start();
			log.info("Start thread " + t.getId());
			// sleep between thread
			try {
				Thread.sleep(Configuration.SEPERATOR);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public static IncomingQueue getInstance(){
		if(q == null)
			q = new IncomingQueue();
		return q;
	}

	public void add2Queue(Incoming in){
		synchronized(queue){
			log.info("New message into IncomingQueue");
			queue.addLast(in);
			queue.notifyAll();
		}
	}

	public void stop(){
		isRunning = false;
	}
	
	private class IncomingWorker extends Thread{
		@Override
		public void run() {
			long threadId = Thread.currentThread().getId();
			while(isRunning){
				long start = System.currentTimeMillis();
				LinkedList<Incoming> myList = new LinkedList<Incoming>();
				synchronized(queue){
					try {
						queue.wait();
					} catch (InterruptedException e) {
						log.error("Error", e);
					}
					log.info("IncomingWorker" + threadId + " check queue");
					int counter = 0;
					while(!queue.isEmpty() && counter < Configuration.MAX_BATCH){
						counter ++;
						myList.add(queue.removeFirst());
					}
				}

				log.info("IncomingWorker" + threadId + " get " + myList.size() + " jobs");
				//TODO: batch update to DB
				if(myList.size() > 0){
					insertRecordsIntoIncoming(myList);
					long end = System.currentTimeMillis();
					long exectime = end - start;
					if(exectime < Configuration.THRESHOLD){
						try {
							Thread.sleep(exectime);
						} catch (InterruptedException e) {
							log.error("Error", e);
						}						
					}
				}
				else{
					try {
						Thread.sleep(Configuration.SLEEP);
					} catch (InterruptedException e) {
						log.error("Error", e);
					}
				}
				log.info("IncomingWorker" + threadId + " finish " + myList.size() + " jobs");					
			}
			log.info("IncomingWorker" + threadId + " stop");
		}
		
		private void insertRecordsIntoIncoming(List<Incoming> listIncomings) {
			log.debug("Insert batch incoming " + listIncomings.size());

			Connection dbConnection = null;
			PreparedStatement preparedStatement = null;

			String insertTableSQL = "insert into incoming (sender,receiver,message,status,session_id,mo_id,unique_id ) values(?,?,?,?,?,?,?)";
			try {
				dbConnection = Database.getConnection();
				preparedStatement = dbConnection.prepareStatement(insertTableSQL);

				dbConnection.setAutoCommit(false);
				for(Incoming incoming : listIncomings){
					preparedStatement.setString(1, incoming.getSender());
					preparedStatement.setString(2, incoming.getReceiver());
					preparedStatement.setString(3, incoming.getMessage());
					preparedStatement.setInt(4, incoming.getStatus());
					preparedStatement.setString(5, incoming.getSession_id());
					preparedStatement.setString(6, incoming.getMo_id());
					preparedStatement.setString(7, incoming.getUnique_id());
					preparedStatement.addBatch();
				}
				preparedStatement.executeBatch();
				dbConnection.commit();
				dbConnection.setAutoCommit(true);
			} catch (SQLException e) {
				log.error("Error", e);
				try {
					dbConnection.rollback();
				} catch (SQLException e1) {
					log.error("Error", e);
				}
			} finally {
				if (preparedStatement != null) {
					try {
						preparedStatement.close();
					} catch (SQLException e) {
						log.error("Error", e);
					}
				}
				if (dbConnection != null) {
					Database.closeConnection(dbConnection);
				}
			}
		}
	}
}
